package com.yunhe.transfer.websocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.transfer.client.MeasHistoryClient;
import com.yunhe.transfer.client.MeasRealtimeClient;
import com.yunhe.common.constant.CacheConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.model.DomainList;
import com.yunhe.common.model.base.DataTransferDTO;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.transfer.util.MeasurementUtil;
import com.yunhe.common.util.RedisClient;
import com.yunhe.transfer.util.WebsocketUtil;
import com.yunhe.common.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.net.URI;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @title: RecollectHandler
 * @Author Chen Lun
 * @Date: 2021/1/22
 */
@Component
public class RecollectHandler extends TextWebSocketHandler {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    private static final Map<String, List<WebSocketSession>> STATION_SESSION_MAP = new ConcurrentHashMap<>();

    private static final Map<WebSocketSession, String> SESSION_STATION_MAP = new ConcurrentHashMap<>();

    @Autowired
    private MeasRealtimeClient measRealtimeClient;
    @Autowired
    private MeasHistoryClient measHistoryClient;
    @Autowired
    private WebsocketUtil websocketUtil;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private RedisClient redisClient;

    private static final Logger logger = LoggerFactory.getLogger(RecollectHandler.class);

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        //保存电站对应session至本地缓存map
        URI sessionUri = session.getUri();
        if (sessionUri != null) {
            List<String> stationCodes = websocketUtil.getStationCodeList(sessionUri.getQuery());
            if (CollectionUtil.isNotEmpty(stationCodes)) {
                for (String stationCode : stationCodes) {
                    List<WebSocketSession> webSocketSessions = STATION_SESSION_MAP.get(stationCode);
                    if (webSocketSessions == null) {
                        webSocketSessions = new ArrayList<>();
                    }
                    webSocketSessions.add(session);
                    STATION_SESSION_MAP.put(stationCode, webSocketSessions);
                    SESSION_STATION_MAP.put(session, stationCode);
                }
            }
        }
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        URI sessionUri = session.getUri();
        if (sessionUri != null) {
            String stationCode = SESSION_STATION_MAP.get(session);
            if (stationCode != null) {
                SESSION_STATION_MAP.remove(session);
                List<WebSocketSession> webSocketSessions = STATION_SESSION_MAP.get(stationCode);
                if (CollectionUtil.isNotEmpty(webSocketSessions) && webSocketSessions != null) {
                    webSocketSessions.remove(session);
                    STATION_SESSION_MAP.put(stationCode, webSocketSessions);
                }
            }
        }
    }

    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        String payload = message.getPayload();
        try {
            logger.info("接收到补录数据", payload);
            Map<String, Object> dataMap = (Map<String, Object>)MAPPER.readValue(payload, Map.class);
            Object listObj = dataMap.get("list");
            Object rtimeObj = dataMap.get("rtime");
            Object stationCode = dataMap.get("stationCode");
            if (listObj != null && stationCode != null) {
                List<Measurement> measurements = MeasurementUtil.map2Meas((List<Map<String, Object>>) listObj, Optional.ofNullable(rtimeObj));
                saveMeas(measurements, Long.parseLong(stationCode.toString()));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void saveMeas(List<Measurement> measurements, Long stationCode) {
        CompletableFuture.supplyAsync(() -> {
            try {
                save(measurements);
                DataTransferDTO dataTransferDTO = MeasurementUtil.recollectData(measurements, stationCode);
                String key = CacheConsts.KEY.RECOLLECT.value(UUID.randomUUID());
                redisClient.setObject(key, dataTransferDTO, CacheConsts.getDefaultTime());
                kafkaTemplate.send(KafkaConsts.Topic.Recollect, key);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        });
    }

    private void save(List<Measurement> measurements) {
//        Map<Long, Double> map = new HashMap<>(measurements.size());
//        for (Measurement measurement : measurements) {
//            map.put(measurement.getPointNumber(), measurement.getVal());
//        }
//        measRealtimeClient.save(map);
        DomainList<Measurement> domainList = new DomainList<>(measurements);
        measRealtimeClient.saveObj(domainList);
        measHistoryClient.save(domainList, MeasurementConsts.FREQUENCY.Original.value());
    }

    /**
     * 订阅指令消息
     * @param dataMap
     * @throws IOException
     */
    public void recollectTopicListenerTest(DataTransferDTO dataMap) throws IOException {
        logger.info("接收到主动补招消息：{}", dataMap);
        Long stationCode = dataMap.getStationCode();
        // 根据电站ID（或电站code）获得对应的ws，并发送数据
        List<WebSocketSession> webSocketSessions = STATION_SESSION_MAP.get(String.valueOf(stationCode));
        // 这里可能需要在ws中加一个映射Map，将电站code和WebSocketSession对应起来
        // 建立连接时（afterConnectionEstablished方法）添加到映射Map
        // 考虑到有多个ws客户端使用同一个电站code的情况，所以是1对多的关系
        if (CollectionUtil.isNotEmpty(webSocketSessions)) {
            webSocketSessions.stream().forEach(webSocketSession -> {
                try {
                    if (!webSocketSession.isOpen()) {
                        webSocketSessions.remove(webSocketSession);
                        SESSION_STATION_MAP.remove(webSocketSession);
                    } else {
                        webSocketSession.sendMessage(new TextMessage(MAPPER.writeValueAsString(dataMap)));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
            STATION_SESSION_MAP.put(String.valueOf(stationCode), webSocketSessions);
        } else {
            logger.error("{}项目连接数为0", stationCode);
        }
    }
}
